package com.etlmaster.datax.plugins.reader.directoryReader;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;

import org.apache.commons.vfs2.FileObject;
import org.apache.log4j.Logger;

import com.dp.nebula.wormhole.common.AbstractPlugin;
import com.dp.nebula.wormhole.common.interfaces.ILine;
import com.dp.nebula.wormhole.common.interfaces.ILineSender;
import com.dp.nebula.wormhole.common.interfaces.IReader;
import com.etlmaster.datax.plugins.tranformer.LineTransFormer;
import com.etlmaster.datax.plugins.writer.vfswriter.VFSFileFinder;

public class DirectoryReader extends AbstractPlugin implements IReader {

	private String filePath;

	private char colDelimiter = '\t';

	private String encoding = "UTF-8";

	private String nullString = "";

	private Logger logger = Logger.getLogger(DirectoryReader.class);

	private FileObject fo = null;

	private InputStream in = null;

	private BufferedReader br = null;

	private int bufferSize = 4 * 1024;
	public static final int LINE_MAX_FIELD = 1024;

	@Override
	public void init() {
		this.filePath = getParam().getValue(ParamKey.directoryPath, "");
		this.nullString = getParam().getValue(ParamKey.nullString, "");
		this.colDelimiter = getParam()
				.getCharValue(ParamKey.colDelimiter, '\t');
		this.encoding = getParam().getValue(ParamKey.encoding, "UTF-8");
		String fileName=filePath.substring(filePath.lastIndexOf("/")+1);
		getParam().putValue("fileName", fileName);
		if(getParam().getValue("changeMap")!=null)
			LineTransFormer.paraseJob(getParam().getValue("changeMap"), getParam());
	}

	@Override
	public void connection() {

		try {
			fo = VFSFileFinder.resolveFile(this.filePath);
			in = fo.getContent().getInputStream();
			br = new BufferedReader(new InputStreamReader(in, encoding),
					bufferSize);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void read(ILineSender lineSender) {
		String lineString = null;
		try {

			while ((lineString = br.readLine()) != null) {
				ILine line = lineSender.createNewLine();
				int i = 0;
				int begin = 0;
				int length = lineString.length();

				for (i = 0; i < length; ++i) {
					if (lineString.charAt(i) == colDelimiter) {
						line.addField(replaceNullString(lineString.substring(
								begin, i)));
						begin = i + 1;
					}
				}
				line.addField(replaceNullString(lineString.substring(begin, i)));

				if(getParam().getValue("changeMap")!=null)
					LineTransFormer.trans(line,getParam());
				if(!line.isSkip()){
					boolean flag = lineSender.send(line);
					if (getMonitor() != null) {
						if (flag) {
							getMonitor().increaseSuccessLines();
						} else {
							getMonitor().increaseFailedLines();
						}
					}
				}
			}
		} catch (IOException e) {
			logger.error(e.getMessage());
		}
		lineSender.flush();

	}

	private String replaceNullString(String string) {
		if (nullString != null && nullString.equals(string)) {
			return null;
		}
		return string;
	}

	@Override
	public void finish() {
		closeAll();
	}

	public void chkFileRows() {

	}

	public void closeAll() {
		try {
			if (br != null) {
				br.close();
			}
			if (in != null) {
				in.close();
			}
			if (fo != null) {
				fo.close();
			}
		} catch (IOException e) {
			logger.error(e.getMessage());
		}
	}

}
